package t_02_nio;

import lombok.AllArgsConstructor;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * https://www.cnblogs.com/jhxxb/p/11303947.html
 * 使用 NIO 完成网络通信的三个核心：
 * 1. 通道（Channel）：负责连接
 * java.nio.channels.Channel 接口：
 * |--SelectableChannel
 * |--SocketChannel
 * |--ServerSocketChannel
 * |--DatagramChannel
 * |--Pipe.SinkChannel
 * |--Pipe.SourceChannel
 * 2. 缓冲区（Buffer）：负责数据的存取
 * 3. 选择器（Selector）：是 SelectableChannel 的多路复用器。用于监控 SelectableChannel 的 IO 状况
 * 可以监听的事件类型（可使用 SelectionKey 的四个常量表示）
 * * 读: SelectionKey.OP_READ （1）
 * * 写: SelectionKey.OP_WRITE （4）
 * * 连接: SelectionKey.OP_CONNECT （8）
 * * 接收: SelectionKey.OP_ACCEPT （16）
 * <p>
 * Selector 常用方法
 * * Set<SelectionKey> keys()：所有的 SelectionKey 集合。代表注册在该 Selector上的 Channel
 * * selectedKeys()：被选择的 SelectionKey 集合。返回此Selector的已选择键集
 * * intselect()：监控所有注册的 Channel，当它们中间有需要处理的 IO 操作时，该方法返回，并将对应得的 SelectionKey 加入被选择的 SelectionKey 集合中，该方法返回这些 Channel 的数量。
 * * int select(long timeout)：可以设置超时时长的 select() 操作
 * * int selectNow()：执行一个立即返回的 select() 操作，该方法不会阻塞线程
 * * Selector wakeup()：使一个还未返回的 select() 方法立即返回
 * * void close()：关闭该选择器
 */
public class NIOServer {
    ExecutorService pool = Executors.newFixedThreadPool(50);
    Selector selector;

    public static void main(String[] args) throws IOException {
        NIOServer server = new NIOServer();
        server.initServer(6666);
        server.listen();
    }

    public void initServer(int port) throws IOException {
        // 创建 ServerSocketChannel -> ServerSocket
        ServerSocketChannel serverChannel = ServerSocketChannel.open();
        // 绑定一个端口, 在服务器端监听
        serverChannel.socket().bind(new InetSocketAddress(port));
        // serverChannel.socket().bind(new InetSocketAddress("127.0.0.1", 8888));
        // 设置为非阻塞
        serverChannel.configureBlocking(false);

        System.out.println("server started, listening on :" + serverChannel.getLocalAddress());
        this.selector = Selector.open(); // 得到一个 Selecor 对象
        // 把 serverSocketChannel 注册到  selector，关心的事件为 OP_ACCEPT
        serverChannel.register(selector, SelectionKey.OP_ACCEPT);
        System.out.println("服务端启动成功！");
    }

    public void listen() throws IOException {
        while (true) { // 循环等待客户端连接
            // selector.select(); // 一直等待连接
            // 等待 10 秒，如果没有事件发生, 返回
            if (selector.select(10000) == 0) { // 没有事件发生
                System.out.println("服务器等待了 10 秒，无连接。当前注册的 key 数量 = " + selector.keys().size());
                System.out.println("服务器等待了 10 秒，无连接。当前注册的 selectionKey 数量 = " + selector.selectedKeys().size());
                continue;
            }

            Set<SelectionKey> selectionKeys = selector.selectedKeys();
            // 如果返回的 > 0, 就获取到相关的 selectionKey 集合，selector.selectedKeys() 返回关注事件的集合
            // 通过 selectionKeys 反向获取通道
            System.out.println("当前注册的 selectionKey 数量 = " + selectionKeys.size());
            // 遍历 Set<SelectionKey>, 使用迭代器遍历
            Iterator<SelectionKey> ite = selectionKeys.iterator();

            while (ite.hasNext()) {
                SelectionKey key = ite.next(); // 获取到 SelectionKey

                // 根据 key 对应的通道发生的事件做相应处理
                if (key.isAcceptable()) { // 如果是 OP_ACCEPT, 有新的客户端连接
                    ServerSocketChannel server = (ServerSocketChannel) key.channel();
                    // 该该客户端生成一个 SocketChannel
                    SocketChannel channel = server.accept();
                    System.out.println("客户端连接成功，生成了一个 socketChannel：" + channel.hashCode());
                    // 将 SocketChannel 设置为非阻塞
                    channel.configureBlocking(false);

                    // new Client
                    String hostIP = ((InetSocketAddress) channel.getRemoteAddress()).getHostString();
                    System.out.println("client " + hostIP + " trying  to connect");
                    // for(int i=0; i<clients.size(); i++) {
                    //     String clientHostIP = clients.get(i).clientAddress.getHostString();
                    // 	if(hostIP.equals(clientHostIP)) {
                    //         System.out.println("this client has already connected! is he alvie " + clients.get(i).live);
                    //         channel.close();
                    // 		return;
                    // 	}
                    // }

                    // 将 socketChannel 注册到 selector, 关注事件为 OP_READ， 同时给 socketChannel 关联一个 Buffer
                    channel.register(selector, SelectionKey.OP_READ, ByteBuffer.allocate(1024));
                    // channel.register(selector, SelectionKey.OP_READ);
                    System.out.println("当前注册的 selectionKey 数量 = " + selector.selectedKeys().size());
                } else if (key.isReadable()) { // 发生 OP_READ
                    key.interestOps(key.interestOps() & (~SelectionKey.OP_READ));
                    pool.execute(new ThreadHandlerChannel(key));
                }

                ite.remove(); // 手动从集合中移除当前的 selectionKey, 防止重复操作
            }
        }
    }


    @AllArgsConstructor
    static class ThreadHandlerChannel extends Thread {
        private final SelectionKey key;

        @Override
        public void run() {
            // 通过 key 反向获取到对应 channel
            SocketChannel channel = (SocketChannel) key.channel();

            try {
                // 获取到该 channel 关联的 buffer
                ByteBuffer byteBuffer = (ByteBuffer) key.attachment();
                channel.read(byteBuffer);
                System.out.println("form 客户端：" + channel.hashCode() + "\t" + new String(byteBuffer.array()));

                ByteBuffer buffer = ByteBuffer.allocate(1024);
                ByteArrayOutputStream baos = new ByteArrayOutputStream();
                int size;
                while ((size = channel.read(buffer)) > 0) {
                    buffer.flip();
                    baos.write(buffer.array(), 0, size);
                    buffer.clear();
                }
                baos.close();
                byte[] content = baos.toByteArray(); // 接收到的内容
                System.out.println("收到消息：" + new String(content));

                // 给客户端发送消息
                ByteBuffer bufferToWrite = ByteBuffer.wrap("HelloClient".getBytes());
                channel.write(bufferToWrite);
                ByteBuffer writeBuf = ByteBuffer.allocate(content.length);
                writeBuf.put(content);
                writeBuf.flip();
                channel.write(writeBuf);

                // 关闭
                if (size == -1) {
                    channel.close();
                } else {
                    key.interestOps(key.interestOps() | SelectionKey.OP_READ);
                    key.selector().wakeup();
                }
            } catch (Exception e) {
                System.out.println(e.getMessage());
            }
        }
    }
}
